package net.techfuser.bigscreen;

import net.techfuser.bigscreen.entity.ChannelStoreSummary;
import net.techfuser.bigscreen.entity.Summary;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-4-6
 * @Version 1.0
 **/
///**
// * Base interface for functions that are evaluated over keyed (grouped) windows.
// *
// * @param <IN> The type of the input value.
// * @param <OUT> The type of the output value.
// * @param <KEY> The type of the key.
// * @param <W> The type of {@code Window} that this window function can be applied on.
// */
//@Public
//public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, Serializable {
class OrderWindow implements WindowFunction<Summary, ChannelStoreSummary, EntRegionChannelKey, TimeWindow> {

    @Override
    public void apply(EntRegionChannelKey keys, TimeWindow window, Iterable<Summary> input, Collector<ChannelStoreSummary> out) throws Exception {
        ChannelStoreSummary result = new ChannelStoreSummary();
        result.ent_id = keys.ent_id;
        result.region_code = keys.region_code;
        result.region_name = keys.region_name;
        result.channel_keyword = keys.channel_keyword;

        Summary summary = input.iterator().next();
        result.count = summary.getCount();
        result.total_sale_value = summary.getTotal_sale_value();

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        result.window_time = String.format("[%s,%s)", sdf.format(window.getStart()), sdf.format(window.getEnd()));
        result.timestamp = sdf.format(System.currentTimeMillis());
//        System.out.println("ChannelStoreSummary:" + JSON.toJSONString(result));

        out.collect(result);
    }
}
